package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/go-gin-admin/pkg/elasticsearch"
	"time"
)

// 处理往kafka写日志

var (
	KafkaClient sarama.SyncProducer // 声明一个全局的连接kafka的生产者client
	logDataChan chan *logData       // 声明一个日志channel
)

// 声明一个日志类型
type logData struct {
	topic string // 主题
	data  string // 数据，json格式
}

// 初始化kafka client
func InitClient(addrs []string) (err error) {
	// 配置kafka
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出⼀个 partition
	config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

	// 连接kafka
	KafkaClient, err = sarama.NewSyncProducer(addrs, config)
	if err != nil {
		fmt.Println("producer init err, ", err)
		return
	}
	fmt.Println("connect kafka success")
	// 初始化logDataChan
	logDataChan = make(chan *logData)
	// 从logDataChan中获取数据发送kafka
	go sendToKafka()
	return
}

// 初始化kafka消费者
func InitConsumer(addrs []string, topic string) (err error)  {
	consumer, err := sarama.NewConsumer(addrs, nil)
	if err != nil {
		fmt.Printf("fail to start consumer, err: %v\n", err)
		return err
	}
	fmt.Println("consumer success")
	partitionList, err := consumer.Partitions(topic) // 获取topic 下所有的 分区
	if err != nil {
		fmt.Printf("fail get list of partition, err: %v\n", err)
		return err
	}
	for partition := range partitionList { // 遍历所有的分区
		// 针对每一个分区创建一个对应的分区消费者
		pc, err := consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
			return err
		}
		// 从每个分区消费消息
		go func(sarama.PartitionConsumer) {
			for msg := range pc.Messages() {
				// 发送到es
				data := elasticsearch.LogData{
					Topic: topic,
					Data:  string(msg.Value),
				}
				elasticsearch.SendToESChan(&data)
			}
		}(pc)
	}
	return err
}


// 向kafka中发送数据
func sendToKafka() {
	for {
		select {
		case ld := <-logDataChan:
			// 构建kafka信息
			message := &sarama.ProducerMessage{}
			message.Topic = ld.topic
			message.Value = sarama.StringEncoder(ld.data)
			// 发送消息 分区和偏移量
			_, _, err := KafkaClient.SendMessage(message)
			if err != nil {
				fmt.Println("send kafka message failed, err: ", err)
				return
			}
			//fmt.Printf("partition: %v offset: %v\n", partition, offset)
		default:
			// 睡一会
			time.Sleep(time.Millisecond * 50)
		}
	}
}

// 暴露函数，将日志数据发送到内部的channel中
func SendToLogDataChan(topic, data string) {
	msg := &logData{
		topic: topic,
		data:  data,
	}
	logDataChan <- msg
}
